package org.spdata.tsdb.client;


import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spdata.tsdb.dto.DataPoint;
import org.spdata.tsdb.dto.Request;
import org.spdata.tsdb.dto.Request.Method;
import org.spdata.utils.SpDataObjectMapper;

import com.fasterxml.jackson.databind.ObjectMapper;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

public class Client extends ChannelInboundHandlerAdapter{
	private static final Logger logger = LoggerFactory.getLogger(Client.class);
	
	private String response ;
	private Object holded = null;
	
	private ChannelHandlerContext ctx;
	private ObjectMapper objMapper = null;
	public enum State{
		send, //发送状态
		idle, //空闲状态
		wait, //等待服务器响应状态
		receive; //接收状态
	}
	private State state = State.idle;
	
	
	protected static Client getInstance(String host,int port) throws Exception{
		Client client = build(host,port);
		return client;
	}
	
	public boolean isClosed () {
		return ctx==null;
	}


	public Object getHolded() {
		return holded;
	}

	public void setHolded(Object holded) {
		this.holded = holded;
	}

	public void release() {
		logger.debug("release a client by {}" , holded);
		this.holded = null;
		
	}

	
	private static Client build(String host,int port)throws Exception{
		final Client client = new Client();
		ChannelFuture f = null;
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		Bootstrap bootstrap = new Bootstrap();
		bootstrap.group(workerGroup);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(new LineBasedFrameDecoder(1024*1024));
            	ch.pipeline().addLast(new StringDecoder());
            	ch.pipeline().addLast(client);
			}
		});

		try {
			f = bootstrap.connect(host, port).sync();
			logger.info("connected tsdb server host={} ,port={}",host,port);
			return client;
		} catch (Exception e) {
			throw new Exception("failed to connect tsdb server host="+host+",port="+port,e);
		}
	}
	
	
	public void channelRead(ChannelHandlerContext ctx, Object msg){	 	
	 	synchronized(this){
	 		this.state = State.receive;
	 		this.response = (String)msg;
	 		logger.debug("recevice msg from server :{}",response);
	 		this.state = State.idle;
	 		this.notify();
	 	}
	}
	
	@Override  
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		synchronized(this){
			logger.error("caught exception of channel! ",cause);
			this.notify();
			try {
				ctx.close();
			}catch(Exception e) {
				logger.error("failed to close ctx!",e);
			}
	        ctx = null;
		}
    }
	
	
	List<DataPoint> shouldPutDps = new ArrayList<DataPoint>();
	
	private Client() {
		objMapper = new SpDataObjectMapper();
	}
	
	public State getState() {
		return state;
	}
	public void setState(State state) {
		this.state = state;
	}
	@Override    
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
		this.ctx = ctx;
	}
	
	public void put(List<DataPoint> dps){
		if(dps!=null){
			this.shouldPutDps.addAll(dps);	
		}
	}
	
	public void put(DataPoint dp){
		if(dp!=null){
			this.shouldPutDps.add(dp);
		}
	}
	
	public void flush() {
		doPut(shouldPutDps);
		this.shouldPutDps.clear();
	}
	
	private void doPut(List<DataPoint> points){			
		String msg = null;
		Request action = new Request();
		action.setData(points);
		action.setMethod(Method.put);
		try{
			msg = objMapper.writeValueAsString(action);		
			logger.debug("do put {} data points.",points.size() );
		}catch(Exception e){
			logger.error("failed to convert dps to String !",e);
		}
		this.put(msg);
	}
	
	public String query(Query query){
		String msg  = null;
		Request action = new Request();
		action.setData(query);
		action.setMethod(Method.query);
		try{
			msg = objMapper.writeValueAsString(action);			
		}catch(Exception e){
			logger.error("failed to convert query to String !",e);
		}
		return this.put(msg);
	}
	
	
	private String put(String requestStr){
		if(requestStr==null)
			return null;
		
		synchronized(this){			
			this.state = State.send;
			byte[] bytes = requestStr.getBytes();
			ByteBuf encoded = ctx.alloc().buffer(bytes.length);    
	        encoded.writeBytes(bytes);    
	        encoded.writeByte('\r');
	        encoded.writeByte('\n');
	        ctx.write(encoded);    
	        ctx.flush();
	        logger.debug("数据已提交");
	        this.state = State.wait;
		
			while(state.equals(State.idle)==false){
				try {
					logger.info("wait for response");
					this.wait(5000);
					break;
				} catch (InterruptedException e) {
					e.printStackTrace();
				}				
			}
			if(state.equals(State.idle)==false) {
				try {
					ctx.close();
				}catch(Exception e) {
					logger.error("failed to close ctx!",e);
				}
		        ctx = null;
				throw new RuntimeException("time out for request");
			}
			if(ctx==null) {
				throw new RuntimeException("current connect has been closed!");
			}else {
				logger.debug("收到返回:{}",response);
				return this.response;
			}
		}
	}
	
	public static void main(String[] args) {
		Date d = new Date(1523447936929L);
		System.out.println(d);
	}
}
